package com.wetsoft.scala

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable
import scala.concurrent.duration._

class Master extends Actor{
  //worker ---->workInfo
  val idToWorker = new mutable.HashMap[String,WorkerInfo]()
  //workerInfo
  val workers = new mutable.HashSet[WorkerInfo]()
  val checkTimeOutWorkerInterval = 5000

  //Master启动之后立即执行的方法
  override def preStart(): Unit = {
    import context.dispatcher
    context.system.scheduler
      .schedule(0 millis,
              checkTimeOutWorkerInterval millis,
              self,
              CheckTimeOutWorker)
  }

  override def receive: Receive = {

    case "started" => println("master startup successful...")
    //接收work发送过来的消息
    case RegisterWorker(workerId, cores, memory) => {
      if(!idToWorker.contains(workerId)){
        //如果不包含workerId,将worker封装成WorkerInfo 保存在idToWorker
        val workerInfo = new WorkerInfo(workerId, cores, memory)
        idToWorker.put(workerId,workerInfo)
        workers += workerInfo
        //4、master向worker发送注册成功的消息给worker
        //sender指代的是消息源，即谁发送过来的消息，就指代谁
        sender() ! RegisteredWorker
      }
    }
    //master接收Worker汇报的心跳
    case SendHeartBeat(workerId) =>{
      println("worker "+workerId+" 向master发送心跳信息.....")
      //从idToWorker中取出workerInfo并且更新最近的心跳
      val workerInfo = idToWorker(workerId)
      workerInfo.lastHeartBeatTime = System.currentTimeMillis()
      workers += workerInfo

    }

    case CheckTimeOutWorker =>{
      val currentTime = System.currentTimeMillis();
      //过滤出超时的worker（即在指定的时间范围内没有向Master进行汇报）
      val toRemoves = workers.filter(w => currentTime - w.lastHeartBeatTime > checkTimeOutWorkerInterval)

      toRemoves.foreach(worker =>{
        idToWorker.remove(worker.workerId)
        workers -= worker
      })
      println(s"worker count is ${workers.size}")
    }
    case shutDownWorker(workerId) =>{
      println("worker["+workerId+"] 停止服务通知Master....")
      val workerInfo  = idToWorker(workerId)
      idToWorker.remove(workerId)
      workers -= workerInfo
    }
  }
}


object Master {

  def main(args: Array[String]): Unit = {

    val masterHost = "localhost"
    val masterPort = "8888"
    /**
      * 在实际应用场景下，有时候我们就是确实需要在scala创建多少字符串，但是每一行需要固定对齐。
      * 解决该问题的方法就是应用scala的stripMargin方法，在scala中stripMargin默认是“|”作为出来连接符，
      * 在多行换行的行头前面加一个“|”符号即可。
      * 当然stripMargin方法也可以自己指定“定界符”,同时更有趣的是利用stripMargin.replaceAll方法，
      * 还可以将多行字符串”合并”一行显示。
      */
    val configStr =
      s"""
         |akka.actor.provider="akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname="$masterHost"
         |akka.remote.netty.tcp.port="$masterPort"
       """.stripMargin

    val config = ConfigFactory.parseString(configStr)
    //创建ActorSystem
    val masterActorSystem = ActorSystem("masterActorSystem", config)
    //使用ActorSystem创建actor
    val masterActor = masterActorSystem.actorOf(Props[Master], "masterActor")
    //给新创建的masterActor发送一条消息，发送消息使用感叹号"!"
    masterActor ! "started"
    //将ActorSystem阻塞在这，不要让其停止
    masterActorSystem.whenTerminated
  }
}
